package com.atguigu.chapter07.D04_State;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * Author: Pepsi
 * Date: 2023/8/8
 * Desc:
 */
public class Flink01_State_Operator_List {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",1000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);

        // 启用checkpoint 周期是2000毫秒，也就是2秒，每隔2s将状态保存一下
        env.enableCheckpointing(2000);

        env
                .socketTextStream("hadoop101",9999)
                .map(new MyMapFunction())
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    // 算子状态，不能用内部类了，因为要实现两个接口，算子状态要实现CheckpointedFunction接口
    private static class MyMapFunction implements MapFunction<String,String>, CheckpointedFunction {

        List<String> words = new ArrayList();
        private ListState<String> wordsState;

        @Override
        public String map(String line) throws Exception {
            if (line.contains("x")) {
                throw new RuntimeException("手动抛出异常...");
            }
            String[] data = line.split(" ");
            words.addAll(Arrays.asList(data));
            return words.toString();
        }

        // 保存状态：周期性的执行
        // 每个并行度都会周期性的执行
        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            // 这个方法是把数据存入到算子状态（状态列表）
//            System.out.println("MyMapFunction.snapshotState");
            //            wordsState.clear();       清空状态
//            wordsState.addAll(words);   向状态中写数据
            // 上面两个方法能用下面这一个方法代替
            wordsState.update(words);
        }

        // 程序启动的时候每个并行度执行一次
        // 这个方法可以把状态中的数据恢复到Java的集合中
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            // 从状态中恢复数据
//            System.out.println("MyMapFunction.initializeState");
            System.out.println("程序重启，从状态中恢复数据...");
            // 获取列表状态
            wordsState = context.getOperatorStateStore().getListState(new ListStateDescriptor<String>("wordsState", String.class));
            // 从列表中获取数据
            // 将状态中的数据遍历出来，在添加到集合中，也就是恢复数据
//            Iterable<String> it = wordsState.get();
            for (String word : wordsState.get()) {
                words.add(word);
            }
        }
    }
}
